package org.bling.star.server.connect.node;

import org.bling.star.common.command.BroadcastCommand;
import org.bling.star.common.connect.CommandStreamKey;
import org.bling.star.common.connect.ConnectClient;
import org.bling.star.common.connect.ConnectStatus;
import org.bling.star.common.response.BaseResponse;
import org.bling.star.server.connect.GlobalPoolManager;
import org.bling.star.server.connect.StreamPool;
import org.bling.star.server.utils.GeneratorUtils;
import io.netty.channel.Channel;
import org.slf4j.Logger;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * 聊天室
 */
public class TalkingStream implements StreamPool {
    private static final Logger log = org.slf4j.LoggerFactory.getLogger(TalkingStream.class);
    private final CommandStreamKey secretKey; //内部生成的
    private final Set<ConnectClient> connectionGroup;
    private boolean isShutdown;
    private final String streamName;

    public TalkingStream() {
        this.secretKey = GeneratorUtils.generateKey();
        this.isShutdown = false;
        this.connectionGroup = new HashSet<>();
        this.streamName = secretKey.getStreamKey();
    }

    @Override
    public boolean refresh(ConnectClient connection) {
        return false;
    }

    @Override
    public boolean disconnect(ConnectClient connection) {
        if (ConnectStatus.isShutdown(connection)) {
            throw new RuntimeException("已断连");
        }
        connectionGroup.remove(connection);
        return false;
    }

    @Override
    public Boolean connect(ConnectClient connection) {
        if (ConnectStatus.isShutdown(connection)) {
            log.error("已断连");
            return false;
        } else if(connectionGroup.contains(connection)){
            log.info("connectionGroup:{}已连接有:{}",connectionGroup,connection.toString());
            return false;
        } else {
            connectionGroup.add(connection);
            return true;
        }
    }

    @Override
    public ConnectClient connect(Channel channel) {
        ConnectClient streamConnection = new InstanceConnection(channel);
        connect(streamConnection);
        return streamConnection;
    }

    @Override
    public boolean shutdown() {
        CountDownLatch closeLatch = new CountDownLatch(connectionGroup.size());
        for (ConnectClient connection : connectionGroup) {
            connection.close();
        }
        if (closeLatch.getCount() == 0) {
            this.isShutdown = true;
        }
        return false;
    }

    @Override
    public Boolean broadcast(String message, ConnectClient serverConnection) {
        BroadcastCommand talkingCommand = new BroadcastCommand();
        talkingCommand.setMessage(message);
        List<ConnectClient> collect = connectionGroup.stream().filter(connect -> !Objects.equals(connect, serverConnection)).collect(Collectors.toList());
        serverConnection.channel().eventLoop().execute(() -> {
                collect.forEach(connection -> {
                    InstanceConnection connection1 = (InstanceConnection)connection;
                    BaseResponse send = connection1.send(talkingCommand);
                    log.info("[发送给用户:{}][发送消息:{}][发送结果:{}]", connection1.getUserInfo().toString(), message, send.getCode());
            });
        });
        serverConnection.send(BaseResponse.success());
        return true;
    }


    @Override
    public Boolean remove(ConnectClient connectClient) {
        boolean remove = connectionGroup.remove(connectClient);
        return remove;
    }

    @Override
    public CommandStreamKey getStreamSecretKey() {
        return this.secretKey;
    }

    public Collection<ConnectClient> getConnectionGroup() {
        return this.connectionGroup;
    }

    public CommandStreamKey getSecretKey() {
        return this.secretKey;
    }

    public boolean isShutdown() {
        return this.isShutdown;
    }

    public String getStreamName() {
        return this.streamName;
    }

    public static class TalkingStreamBuilder {
        private static TalkingStream build() {
            TalkingStream talkingStream = new TalkingStream();
            GlobalPoolManager.addPool(talkingStream);
            log.info("[新建聊天室:{}]",talkingStream.toString());
            return talkingStream;
        }
    }

    public static TalkingStream build() {
        return TalkingStreamBuilder.build();
    }
}
